package com.doit.day01;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * 创建一个topic user_info
 * 写一个消费者来消费这个topic user_info 的数据
 * 将数据写入到mysql中
 */

public class ConsumerDemo {
    public static void main(String[] args) {
        Properties props = new Properties();
        //必配项
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"group03");

        //选配的
        //auto.offset.reset  指定从哪个地方开始读取数据
        /*
        earliest:最早的
        latest:最晚的
         */
        /**
         * 如果这个组id没有在__consumer_offset 中记录过任何的消息
         * 那么他去__consumer_offset 这个topic中找不到
         * 就回去看，AUTO_OFFSET_RESET_CONFIG,"earliest"  在代码中有没有设置，如果有
         * 就按照你设置的为准
         *
         * 组id没有在__consumer_offset 中有过记录
         * 就直接会读取相应的数据，看上一次消费到了哪里
         * 然后紧接着上次消费的地方开始读
         *
         * 1.从最开始读取  earliest==》__consumer_offset 中不能够有记录
         * 2.从最新的位置开始读   latest==》__consumer_offset 中不能够有记录
         * 3.从上次消费到的位置开始读取数据    __consumer_offset 有记录 就会从有记录的位置
         * 4.从我自己指定的位置开始读  告诉他哪一个topic的哪一个分区的哪一个offset
         */
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        //allow.auto.create.topics   运行自动创建topic
        props.setProperty(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG,"true");

        //enable.auto.commit   运行自动提交偏移量  记录消费到哪了    __consumer_offset
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");
        //auto.commit.interval.ms  每间隔多久自动提交一次，默认是5s
        props.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"10000");
        //创建消费者对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        //拉取的数据是哪一个topic里面的
        //订阅主题
        consumer.subscribe(Arrays.asList("doit40"));

        while (true){
            //拉取数据
            ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(Integer.MAX_VALUE));

            for (ConsumerRecord<String, String> record : poll) {
                String topic = record.topic();
                int partition = record.partition();
                long offset = record.offset();
                String value = record.value();
                TimestampType timestampType = record.timestampType();
                long timestamp = record.timestamp();
                System.out.println("topic:"+topic+",partition:"+partition+",offset:"+offset
                +",value:"+value+",时间戳类型："+timestampType+",时间戳是："+timestamp);
            }
        }
    }
}
